package com.tcxhb.job.core.schedule;

import com.tcxhb.job.core.constants.enums.LogActionEnum;
import com.tcxhb.job.core.utils.LogUtils;
import com.tcxhb.job.dao.dataobject.JobInstanceDO;
import com.tcxhb.job.dao.dataobject.JobMachineDO;
import com.tcxhb.job.dao.dataobject.query.JobInstanceQuery;
import com.tcxhb.job.common.Thread.NamedThreadFactory;
import com.tcxhb.job.common.constants.CommonErrorCode;
import com.tcxhb.job.common.constants.JobApiConstants;
import com.tcxhb.job.common.constants.enums.ExecuteStatus;
import com.tcxhb.job.common.constants.enums.ScheduleStatus;
import com.tcxhb.job.common.model.MiscResult;
import com.tcxhb.job.common.model.PageRequest;
import com.tcxhb.job.common.utils.JsonUtils;
import com.tcxhb.job.core.constants.enums.OnlineStatus;
import com.tcxhb.job.core.service.job.JobInstanceService;
import com.tcxhb.job.core.service.job.JobMachineService;
import com.tcxhb.job.core.utils.UrlUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 任务实例状态扫描; 异步多线程,没毛病
 * @Auther: tcxhb
 * @Date: 2023/11/13
 */
@Component
@Slf4j
public class JobStatusScheduler extends BaseScheduler implements Scheduler {
    @Autowired
    private JobInstanceService jobInstanceService;
    @Autowired
    private JobMachineService machineService;
    @Autowired
    private CloseableHttpAsyncClient asyncClient;
    private ScheduledExecutorService jobBeatSchedule = Executors.newScheduledThreadPool(1,
            new NamedThreadFactory("job-status", true));
    // 间隔时间2分钟获取一次状态
    private Integer periodTime = 120;

    @Override
    @PostConstruct
    public void start() {
        jobBeatSchedule.scheduleAtFixedRate(() -> {
            try {
                log.info("job-status-check-start");
                scan();
            } catch (Exception e) {
                log.error("agent-check-schedule-exp", e);
            }
        }, 61, periodTime, TimeUnit.SECONDS);
    }

    /**
     * 扫描
     */
    public void scan() {
        Long last = 0L;
        JobInstanceQuery query = buildQuery();
        while (true) {
            query.setLastId(last);
            List<JobInstanceDO> list = jobInstanceService.scan(query);
            if (CollectionUtils.isEmpty(list)) {
                break;
            }
            //遍历所有机器
            for (JobInstanceDO instanceDO : list) {
                statusCheck(instanceDO);
                last = instanceDO.getId();
            }
        }
    }

    public void statusCheck(JobInstanceDO instanceDO) {
        String url = instanceDO.getMachineAddr() + JobApiConstants.cmd_status;
        HttpGet get = new HttpGet(url + "?instanceId=" + instanceDO.getId());
        try {
            asyncClient.execute(get, new FutureCallback<HttpResponse>() {
                @Override
                public void completed(HttpResponse response) {
                    try {
                        String str = EntityUtils.toString(response.getEntity());
                        MiscResult<Integer> result = JsonUtils.toMiscBean(str, Integer.class);
                        callback(instanceDO, result);
                    } catch (IOException e) {
                        log.error("completed exp", e);
                    }
                }

                @Override
                public void failed(Exception e) {
                    try {
                        log.error("agent-check-exp,incId:" + instanceDO.getId() + "," + url, e);
                        //机器任务不在线
                        String[] port = UrlUtils.splitHttp(instanceDO.getMachineAddr());
                        JobMachineDO machineDO = machineService.queryByHost(port[0], Integer.valueOf(port[1]), instanceDO.getAppName());
                        if (machineDO == null || OnlineStatus.offline.eq(machineDO.getStatus())) {
                            callback(instanceDO, MiscResult.err(CommonErrorCode.HTTP_TO_AGENT_FAIL));
                        }
                    } catch (Exception ex) {
                        log.error("faild", ex);
                    }
                }

                @Override
                public void cancelled() {
                }
            });
        } catch (Exception e) {
            log.error("agent-check-exp" + url, e);
        }
    }

    /**
     * 状态回调
     *
     * @param instanceDO
     * @param result
     */
    private void callback(JobInstanceDO instanceDO, MiscResult<Integer> result) {
        try {
            //更新状态
            JobInstanceDO updateDO = new JobInstanceDO();
            updateDO.setId(instanceDO.getId());
            // 任务改成失败
            if (!result.isSuccess()) {
                updateDO.setExecuteStatus(ExecuteStatus.FAIL.getType());
                updateDO.setMsg(result.getCode());
            } else if (result.getData() == null) {
                updateDO.setExecuteStatus(ExecuteStatus.FAIL.getType());
                updateDO.setMsg("agent-status-not-found");
            } else {
                updateDO.setExecuteStatus(result.getData());
                updateDO.setMsg("server-check-status");
            }
            //更新数据
            LogUtils.info(instanceDO.getTraceId(), LogActionEnum.CheckInstanceStatus,updateDO.getMsg()+","+updateDO.getExecuteStatus());
            jobInstanceService.updateById(updateDO);
        } catch (Exception e) {
            log.error("instance-status-sch-exp:" + instanceDO.getMachineAddr(), e);
        }
    }

    /**
     * @return
     */
    private JobInstanceQuery buildQuery() {
        JobInstanceQuery query = new JobInstanceQuery();
        query.setPage(new PageRequest());
        query.getPage().setPageNum(1000);
        //调度成功的;属于running状态的
        query.setScheduleStatus(ScheduleStatus.SUCC.getType());
        query.setMinExeStatus(ExecuteStatus.INIT.getType());
        query.setMaxExeStatus(ExecuteStatus.RUNNING.getType());
        return query;
    }
}
